#region Copyright and Licensing Notice
// Copyright (C) 2011 Steven Bosnick
// 
// This program is open source software; you can redistribute it 
// and/or modifiy it under the terms of the Eclipse Public License,
// version 1.0.
// 
// This program is distributed in the hope that it will be useful, 
// but WITHOUT ANY WARRANTY; without even the implied
// warranty of MERCHANTABILITY or FITNESS FOR A 
// PARTICULAR PURPOSE. See the Eclipse Public License for more
// details.
// 
// You should have received a copy of the Eclipse Public License 
// along with this program; if not, you can find a copy at
// "http://www.opensource.org/licenses/eclipse-1.0.php".
#endregion
using System;
using System.IO;
using BitwiseOre.IO;
using BitwiseOre.Properties;
using BitwiseOre.SharedMemory.Core;
using BitwiseOre.SharedMemory.JetBrains;
using BitwiseOre.SharedMemory.Utils;

namespace BitwiseOre.SharedMemory
{
    /// <summary>
    /// Provides one side of a two-way interprocess communicaition that uses shared memory.
    /// </summary>
    [UsedImplicitly(ImplicitUseTargetFlags.WithMembers)]
    public sealed class SharedMemoryStream : DisconnectableStream
    {
        private static readonly string UnableToWriteMessage = Resources.SharedMemoryStream_UnableToWriteMessage;
        private static readonly string UnableToReadMessage = Resources.SharedMemoryStream_UnableToReadMessage;
        private static readonly string UnableToDisconnectMessage = Resources.SharedMemoryStream_UnableToDisconnectMessage;

        private static readonly int DefaultReadTimeout = (int) TimeSpan.FromMinutes(1).TotalMilliseconds;
        private static readonly int DefaultWriteTimeout = (int) TimeSpan.FromMinutes(1).TotalMilliseconds;
        private static readonly int DefaultDisconnectTimeout = (int) TimeSpan.FromMinutes(1).TotalMilliseconds;

        private readonly IDisposer<SharedMemoryStream> _disposer;
        private readonly IWriter _writer;
        private readonly IReader _reader;
        private readonly IDisconnector _disconnector;
        private readonly IAsyncResultEnder _asyncEnder;
        private readonly IAsyncResultEnder<int> _intAsyncEnder;

        internal SharedMemoryStream(IDisposer<SharedMemoryStream> disposer, IWriter writer, IReader reader,
                                    IDisconnector disconnector, IAsyncResultEnder asyncEnder,
                                    IAsyncResultEnder<int> intAsyncEnder)
        {
            _disposer = disposer;
            _writer = writer;
            _reader = reader;
            _disconnector = disconnector;
            _asyncEnder = asyncEnder;
            _intAsyncEnder = intAsyncEnder;

            ReadTimeout = DefaultReadTimeout;
            WriteTimeout = DefaultWriteTimeout;
            DisconnectTimeout = DefaultDisconnectTimeout;
        }

        /// <summary>
        /// Creates connection to a <see cref="SharedMemoryListener"/> of a given name.
        /// </summary>
        /// <param name="name">
        ///     The name of the <see cref="SharedMemoryListener"/> to which to connect.
        /// </param>
        /// <returns>
        ///     A <see cref="SharedMemoryStream"/> that represents one side
        ///     of the interprocess communication channel established with the
        ///     named listner.
        /// </returns>
        /// <exception cref="ListenerNotFoundException">
        ///     No <see cref="SharedMemoryListener"/> named <paramref name="name"/> could
        ///     be found.
        /// </exception>
        /// <exception cref="ArgumentNullException">
        ///     <paramref name="name"/> is <c>null</c>.
        /// </exception>
        public static DisconnectableStream Connect(string name)
        {
            using (var factory = SharedMemoryFactory.Create(name))
            {
                return factory.Connect();
            }
        }

        #region Dispose and Flush

        /// <summary>
        /// Releases the unmanaged resources used by the <see cref="T:System.IO.Stream"/> 
        /// and optionally releases the managed resources.
        /// </summary>
        /// <param name="disposing">
        ///     true to release both managed and unmanaged resources; false to release only unmanaged resources.
        /// </param>
        protected override void Dispose(bool disposing)
        {
            if (disposing)
            {
                _disposer.Dispose();
            }
        }

        /// <summary>
        /// Clears all buffers for this stream and causes any buffered data to be written to the underlying device.
        /// </summary>
        /// <exception cref="T:System.IO.IOException">
        ///     An I/O error occurs.</exception>
        public override void Flush()
        {
            _disposer.ThrowIfDisposed();

            _writer.Flush();
            _reader.Flush();
        }

        #endregion

        #region Disconnect Operations

        /// <summary>
        /// Performs an orderly disconnection of the two sides of the interprocess
        /// communication channel.
        /// </summary>
        /// <exception cref="SharedMemoryException">
        ///     An I/O error occured while disconnecting the stream. This is not an expected occurance.. 
        /// </exception>
        /// <exception cref="T:System.ObjectDisposedException">
        ///     Methods were called after the stream was closed. 
        /// </exception>
        /// <exception cref="TimeoutException">
        ///     The operation timed out.
        /// </exception>
        public override void Disconnect()
        {
            _disposer.ThrowIfDisposed();

            var disconnect = _disconnector.Disconnect(TimeSpan.FromMilliseconds(DisconnectTimeout));
            disconnect.RunSynchronously();

            if (disconnect.Exception != null)
            {
                throw disconnect.Exception.Wrap(CreateException(UnableToDisconnectMessage));
            }
        }

        /// <summary>
        /// Begins an asychronous operation to perform an order disconnection of the
        /// two sides on the interprocess communication channel.
        /// </summary>
        /// <param name="callback">
        ///     An optional asynchronous callback, to be called when the disconnection is complete. 
        /// </param>
        /// <param name="state">
        ///     A user-provided object that distinguishes this particular asynchronous disconnect request
        ///     from other requests. 
        /// </param>
        /// <returns>
        ///    An <see cref="IAsyncResult"/> that represents this asynchonous operation.
        /// </returns>
        /// <exception cref="T:System.ObjectDisposedException">
        ///     Methods were called after the stream was closed. 
        /// </exception>
        public override IAsyncResult BeginDisconnect(AsyncCallback callback, object state)
        {
            _disposer.ThrowIfDisposed();

            var task = _disconnector.Disconnect(TimeSpan.FromMilliseconds(DisconnectTimeout));
            task.Start();

            return task.WithAsyncCallback(callback, state);
        }

        /// <summary>
        /// Completes an asynchronous disconnection operation.
        /// </summary>
        /// <param name="asyncResult">
        ///     The asynchronous operation to complete. Must be the return value
        ///     from <see cref="BeginDisconnect"/>.
        /// </param>
        /// <exception cref="T:System.ArgumentNullException">
        ///     <paramref name="asyncResult"/> is null. 
        /// </exception>
        /// <exception cref="T:System.ArgumentException">
        ///     <paramref name="asyncResult"/> did not originate from a <see cref="BeginDisconnect"/> method 
        ///     on the current stream. 
        /// </exception>
        /// <exception cref="SharedMemoryException">
        ///     An I/O error occured while disconnecting the stream. This is not an expected occurance.. 
        /// </exception>
        /// <exception cref="T:System.ObjectDisposedException">
        ///     Methods were called after the stream was closed. 
        /// </exception>
        /// <exception cref="TimeoutException">
        ///     The operation timed out.
        /// </exception>
        public override void EndDisconnect(IAsyncResult asyncResult)
        {
            _disposer.ThrowIfDisposed();

            _asyncEnder.EndTask(asyncResult, CreateException(UnableToDisconnectMessage));
        }

        #endregion

        #region Read Operations

        /// <summary>
        /// Reads a sequence of bytes from the current stream and advances the position within the stream 
        /// by the number of bytes read.
        /// </summary>
        /// <returns>
        ///     The total number of bytes read into the buffer. This can be less than the number of bytes 
        ///     requested if that many bytes are not currently available, or zero (0) if the end of the 
        ///     stream has been reached.
        /// </returns>
        /// <param name="buffer">
        ///     An array of bytes. When this method returns, the buffer contains the specified byte array 
        ///     with the values between <paramref name="offset"/> and (<paramref name="offset"/> + 
        ///     <paramref name="count"/> - 1) replaced by the bytes read from the current source. 
        /// </param>
        /// <param name="offset">
        ///     The zero-based byte offset in <paramref name="buffer"/> at which to begin storing the 
        ///     data read from the current stream. 
        /// </param>
        /// <param name="count">
        ///     The maximum number of bytes to be read from the current stream. 
        /// </param>
        /// <exception cref="T:System.ArgumentException">
        ///     The sum of <paramref name="offset"/> and <paramref name="count"/> is larger than the 
        ///     buffer length. 
        /// </exception>
        /// <exception cref="T:System.ArgumentNullException">
        ///     <paramref name="buffer"/> is null. 
        /// </exception>
        /// <exception cref="T:System.ArgumentOutOfRangeException">
        ///     <paramref name="offset"/> or <paramref name="count"/> is negative. 
        /// </exception>
        /// <exception cref="SharedMemoryException">
        ///     An I/O error occured while reading from the stream. This is not an expected occurance.. 
        /// </exception>
        /// <exception cref="T:System.ObjectDisposedException">
        ///     Methods were called after the stream was closed. 
        /// </exception>
        /// <exception cref="TimeoutException">
        ///     The operation timed out.
        /// </exception>
        public override int Read(byte[] buffer, int offset, int count)
        {
            _disposer.ThrowIfDisposed();

            try
            {
                return _reader.Read(new ArraySegment<byte>(buffer, offset, count),
                                    TimeSpan.FromMilliseconds(ReadTimeout));
            }
            catch(InvalidOperationException e)
            {
                throw new SharedMemoryException(UnableToReadMessage, e);
            }
            catch (AggregateException e)
            {
                throw e.Wrap(CreateException(UnableToReadMessage));
            }
        }

        /// <summary>
        /// Reads a byte from the stream and advances the position within the stream by one byte, or returns -1 if at the end of the stream.
        /// </summary>
        /// <returns>
        ///     The unsigned byte cast to an Int32, or -1 if at the end of the stream.
        /// </returns>
        /// <exception cref="T:System.ObjectDisposedException">
        ///     Methods were called after the stream was closed. 
        /// </exception>
        /// <exception cref="TimeoutException">
        ///     The operation timed out.
        /// </exception>
        /// <exception cref="SharedMemoryException">
        ///     An unexpected IO error occured while reading.
        /// </exception>
        public override int ReadByte()
        {
            _disposer.ThrowIfDisposed();

            try
            {
                return _reader.ReadByte(TimeSpan.FromMilliseconds(ReadTimeout));
            }
            catch (InvalidOperationException e)
            {
                throw new SharedMemoryException(UnableToReadMessage, e);
            }
            catch (AggregateException e)
            {
                throw e.Wrap(CreateException(UnableToReadMessage));
            }
        }

        /// <summary>
        /// Begins an asynchronous read operation.
        /// </summary>
        /// <returns>
        ///     An <see cref="T:System.IAsyncResult"/> that represents the asynchronous read, which could 
        ///     still be pending.
        /// </returns>
        /// <param name="buffer">
        ///     The buffer to read the data into. 
        /// </param>
        /// <param name="offset">
        ///     The byte offset in <paramref name="buffer"/> at which to begin writing data read from the stream. 
        /// </param>
        /// <param name="count">
        ///     The maximum number of bytes to read. 
        /// </param>
        /// <param name="callback">
        ///     An optional asynchronous callback, to be called when the read is complete. 
        /// </param>
        /// <param name="state">
        ///     A user-provided object that distinguishes this particular asynchronous read request from 
        ///     other requests. 
        /// </param>
        /// <exception cref="T:System.ArgumentException">
        ///     The sum of <paramref name="offset"/> and <paramref name="count"/> is larger than the 
        ///     buffer length. 
        /// </exception>
        /// <exception cref="T:System.ArgumentNullException">
        ///     <paramref name="buffer"/> is null. 
        /// </exception>
        /// <exception cref="T:System.ArgumentOutOfRangeException">
        ///     <paramref name="offset"/> or <paramref name="count"/> is negative. 
        /// </exception>
        /// <exception cref="T:System.ObjectDisposedException">
        ///     Methods were called after the stream was closed. 
        /// </exception>
        public override IAsyncResult BeginRead(byte[] buffer, int offset, int count,
                                               AsyncCallback callback, object state)
        {
            _disposer.ThrowIfDisposed();

            return _reader.ReadAsync(new ArraySegment<byte>(buffer, offset, count),
                                     TimeSpan.FromMilliseconds(ReadTimeout))
                .WithAsyncCallback(callback, state);
        }

        /// <summary>
        /// Waits for the pending asynchronous read to complete.
        /// </summary>
        /// <returns>
        ///     The number of bytes read from the stream, between zero (0) and the number of bytes you 
        ///     requested. Streams return zero (0) only at the end of the stream, otherwise, they should 
        ///     block until at least one byte is available.
        /// </returns>
        /// <param name="asyncResult">
        ///     The reference to the pending asynchronous request to finish. 
        /// </param>
        /// <exception cref="T:System.ArgumentNullException">
        ///     <paramref name="asyncResult"/> is null. 
        /// </exception>
        /// <exception cref="T:System.ArgumentException">
        ///     <paramref name="asyncResult"/> did not originate from a <see cref="BeginRead"/> method 
        ///     on the current stream. 
        /// </exception>
        /// <exception cref="SharedMemoryException">
        ///     An I/O error occured while reading from the stream. This is not an expected occurance.. 
        /// </exception>
        /// <exception cref="T:System.ObjectDisposedException">
        ///     Methods were called after the stream was closed. 
        /// </exception>
        /// <exception cref="TimeoutException">
        ///     The operation timed out.
        /// </exception>
        public override int EndRead(IAsyncResult asyncResult)
        {
            _disposer.ThrowIfDisposed();

            return _intAsyncEnder.EndTask(
                asyncResult,
                CreateException(UnableToReadMessage));
        }

        #endregion

        #region Write Operations

        /// <summary>
        /// Writes a sequence of bytes to the current stream and advances the current position within this 
        /// stream by the number of bytes written.
        /// </summary>
        /// <param name="buffer">
        ///     An array of bytes. This method copies <paramref name="count"/> bytes from 
        ///     <paramref name="buffer"/> to the current stream. 
        /// </param>
        /// <param name="offset">
        ///     The zero-based byte offset in <paramref name="buffer"/> at which to begin copying bytes to 
        ///     the current stream. 
        /// </param>
        /// <param name="count">
        ///     The number of bytes to be written to the current stream. 
        /// </param>
        /// <exception cref="T:System.ArgumentException">
        ///     The sum of <paramref name="offset"/> and <paramref name="count"/> is greater than the 
        ///     buffer length. 
        /// </exception>
        /// <exception cref="T:System.ArgumentNullException">
        ///     <paramref name="buffer"/> is null. 
        /// </exception>
        /// <exception cref="T:System.ArgumentOutOfRangeException">
        ///     <paramref name="offset"/> or <paramref name="count"/> is negative. 
        /// </exception>
        /// <exception cref="T:System.ObjectDisposedException">
        ///     Methods were called after the stream was closed. 
        /// </exception>
        /// <exception cref="TimeoutException">
        ///     The operation timed out.
        /// </exception>
        /// <exception cref="SharedMemoryException">
        ///     An unexpected IO error occured while writing.
        /// </exception>
        public override void Write(byte[] buffer, int offset, int count)
        {
            _disposer.ThrowIfDisposed();

            var task = _writer.Write(new ArraySegment<byte>(buffer, offset, count),
                                     TimeSpan.FromMilliseconds(WriteTimeout));
            task.RunSynchronously();

            if (task.Exception != null)
            {
                throw task.Exception.Wrap(CreateException(UnableToWriteMessage));
            }
        }

        /// <summary>
        /// Writes a byte to the current position in the stream and advances the position within the 
        /// stream by one byte.
        /// </summary>
        /// <param name="value">
        ///     The byte to write to the stream. 
        /// </param>
        /// <exception cref="T:System.ObjectDisposedException">
        ///     Methods were called after the stream was closed.
        /// </exception>
        /// <exception cref="TimeoutException">
        ///     The operation timed out.
        /// </exception>
        /// <exception cref="SharedMemoryException">
        ///     An unexpected IO error occured while writing.
        /// </exception>
        public override void WriteByte(byte value)
        {
            _disposer.ThrowIfDisposed();

            try
            {
                _writer.WriteByte(value, TimeSpan.FromMilliseconds(WriteTimeout));
            }
            catch (AggregateException e)
            {
                throw e.Wrap(CreateException(UnableToWriteMessage));
            }
        }

        /// <summary>
        /// Begins an asynchronous write operation.
        /// </summary>
        /// <returns>
        ///     An IAsyncResult that represents the asynchronous write, which could still be pending.
        /// </returns>
        /// <param name="buffer">
        ///     The buffer to write data from. 
        /// </param>
        /// <param name="offset">
        ///     The byte offset in <paramref name="buffer"/> from which to begin writing. 
        /// </param>
        /// <param name="count">
        ///     The maximum number of bytes to write. 
        /// </param>
        /// <param name="callback">
        ///     An optional asynchronous callback, to be called when the write is complete. 
        /// </param>
        /// <param name="state">
        ///     A user-provided object that distinguishes this particular asynchronous write request from 
        ///     other requests. 
        /// </param>
        /// <exception cref="T:System.ArgumentException">
        ///     The sum of <paramref name="offset"/> and <paramref name="count"/> is greater than the 
        ///     buffer length. 
        /// </exception>
        /// <exception cref="T:System.ArgumentNullException">
        ///     <paramref name="buffer"/> is null. 
        /// </exception>
        /// <exception cref="T:System.ArgumentOutOfRangeException">
        ///     <paramref name="offset"/> or <paramref name="count"/> is negative. 
        /// </exception>
        /// <exception cref="T:System.ObjectDisposedException">
        ///     Methods were called after the stream was closed. 
        /// </exception>
        public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count,
                                                AsyncCallback callback, object state)
        {
            _disposer.ThrowIfDisposed();

            var task = _writer.Write(new ArraySegment<byte>(buffer, offset, count),
                                     TimeSpan.FromMilliseconds(WriteTimeout));
            task.Start();

            return task.WithAsyncCallback(callback, state);
        }

        /// <summary>
        /// Ends an asynchronous write operation.
        /// </summary>
        /// <param name="asyncResult">
        ///     A reference to the outstanding asynchronous I/O request. 
        /// </param>
        /// <exception cref="T:System.ArgumentNullException">
        ///     <paramref name="asyncResult"/> is null. 
        /// </exception>
        /// <exception cref="T:System.ArgumentException">
        ///     <paramref name="asyncResult"/> did not originate from a <see cref="BeginWrite"/>
        ///     method on the current stream. 
        /// </exception>
        /// <exception cref="SharedMemoryException">
        ///     An internal error has occurred.
        /// </exception>
        /// <exception cref="T:System.ObjectDisposedException">
        ///     Methods were called after the stream was closed. 
        /// </exception>
        /// <exception cref="TimeoutException">
        ///     The operation timed out.
        /// </exception>
        public override void EndWrite(IAsyncResult asyncResult)
        {
            _disposer.ThrowIfDisposed();

            _asyncEnder.EndTask(asyncResult, CreateException(UnableToWriteMessage));
        }

        #endregion

        #region Capabilities and Timeouts

        /// <summary>
        /// Gets a value indicating whether the current stream supports seeking.
        /// </summary>
        /// <returns>
        ///     <c>false</c> because the stream does not support seeking.
        /// </returns>
        public override bool CanSeek
        {
            get { return false; }
        }

        /// <summary>
        /// Gets a value indicating whether the current stream supports reading.
        /// </summary>
        /// <returns>
        ///     <c>true</c> if the stream supports reading; otherwise, <c>false</c>
        /// </returns>
        /// <remarks>
        ///     Returns <c>true</c> unless the stream has been closed or disposed.
        /// </remarks>
        public override bool CanRead
        {
            get { return !_disposer.IsDisposed; }
        }

        /// <summary>
        /// Gets a value indicating whether the current stream supports writing.
        /// </summary>
        /// <returns>
        ///     <c>true</c> if the stream supports writing; otherwise, <c>false</c>
        /// </returns>
        /// <remarks>
        ///     Returns <c>true</c> unless the stream has been closed or disposed.
        /// </remarks>
        public override bool CanWrite
        {
            get { return !_disposer.IsDisposed; }
        }

        /// <summary>
        /// Gets a value that determines whether the current stream can time out.
        /// </summary>
        /// <returns>
        ///     <c>true</c> because the stream can time out.
        /// </returns>
        public override bool CanTimeout
        {
            get { return true; }
        }

        /// <summary>
        /// Gets or sets a value, in miliseconds, that determines how long the stream will attempt to read 
        /// before timing out. 
        /// </summary>
        /// <returns>
        ///     A value, in miliseconds, that determines how long the stream will attempt to read before
        ///     timing out.
        /// </returns>
        public override int ReadTimeout{ get; set; }

        /// <summary>
        /// Gets or sets a value, in miliseconds, that determines how long the stream will attempt to write 
        /// before timing out. 
        /// </summary>
        /// <returns>
        ///     A value, in miliseconds, that determines how long the stream will attempt to write before 
        ///     timing out.
        /// </returns>
        public override int WriteTimeout { get; set; }

        /// <summary>
        /// Gets or sets a value, in millisecond, that determines how long the stream will attempt to 
        /// disconnect before timing out.
        /// </summary>
        /// <returns>
        ///     A value, in miliseconds, that determines how long the stream will attempt to disconnect before
        ///     timing out.
        /// </returns>
        public override int DisconnectTimeout { get; set; }

        #endregion

        #region Not Supported Because CanSeek is False

        /// <summary>
        /// Sets the position within the current stream.
        /// </summary>
        /// <returns>
        ///     The new position within the current stream.
        /// </returns>
        /// <param name="offset">
        ///     A byte offset relative to the <paramref name="origin"/> parameter. 
        /// </param>
        /// <param name="origin">
        ///     A value of type <see cref="T:System.IO.SeekOrigin"/> indicating the reference point used to 
        ///     obtain the new position. 
        /// </param>
        /// <exception cref="T:System.NotSupportedException">
        ///     The stream does not support seeking. 
        /// </exception>
        /// <remarks>
        ///     Always throws <seealso cref="NotSupportedException"/> because <seealso cref="CanSeek"/>
        ///     is <c>false</c>.
        /// </remarks>
        public override long Seek(long offset, SeekOrigin origin)
        {
            throw new NotSupportedException();
        }

        /// <summary>
        /// Sets the length of the current stream.
        /// </summary>
        /// <param name="value">
        ///     The desired length of the current stream in bytes. 
        /// </param>
        /// <exception cref="T:System.IO.IOException">An I/O error occurs. </exception>
        /// <exception cref="T:System.NotSupportedException">
        ///     The stream does not support both writing and seeking.
        /// </exception>
        /// <remarks>
        ///     Always throws <seealso cref="NotSupportedException"/> because <seealso cref="CanSeek"/>
        ///     is <c>false</c>.
        /// </remarks>
        public override void SetLength(long value)
        {
            throw new NotSupportedException();
        }

        /// <summary>
        /// Gets the length in bytes of the stream.
        /// </summary>
        /// <returns>
        /// A long value representing the length of the stream in bytes.
        /// </returns>
        /// <exception cref="T:System.NotSupportedException">
        ///     A class derived from Stream does not support seeking. 
        /// </exception>
        /// <remarks>
        ///     Always throws <seealso cref="NotSupportedException"/> because <seealso cref="CanSeek"/>
        ///     is <c>false</c>.
        /// </remarks>
        public override long Length
        {
            get { throw new NotSupportedException(); }
        }

        /// <summary>
        /// Gets or sets the position within the current stream.
        /// </summary>
        /// <returns>
        ///     The current position within the stream.
        /// </returns>
        /// <exception cref="T:System.NotSupportedException">
        ///     The stream does not support seeking. 
        /// </exception>
        /// <remarks>
        ///     Always throws <seealso cref="NotSupportedException"/> because <seealso cref="CanSeek"/>
        ///     is <c>false</c>.
        /// </remarks>
        public override long Position
        {
            get { throw new NotSupportedException(); }
            set { throw new NotSupportedException(); }
        }

        #endregion

        private Func<AggregateException, Exception> CreateException(string message)
        {
            return ae => new SharedMemoryException(message, ae);
        }
    }
}